package com.kiss.rpc.channelhandler.handler;

import com.kiss.rpc.RpcBootstrap;
import com.kiss.rpc.ServiceConfig;
import com.kiss.rpc.enumeration.RequestType;
import com.kiss.rpc.enumeration.ResponseCode;
import com.kiss.rpc.protection.RateLimiter;
import com.kiss.rpc.protection.TokenBuketRateLimiter;
import com.kiss.rpc.transport.message.RequestPayload;
import com.kiss.rpc.transport.message.RpcRequest;
import com.kiss.rpc.transport.message.RpcResponse;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.Method;
import java.net.SocketAddress;
import java.util.Map;

@Slf4j
public class MethodCallHandler extends SimpleChannelInboundHandler<RpcRequest> {

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, RpcRequest rpcRequest){
        // 1、先封装部分响应
        RpcResponse rpcResponse = new RpcResponse();
        rpcResponse.setRequestId(rpcRequest.getRequestId());
        rpcResponse.setCompressType(rpcRequest.getCompressType());
        rpcResponse.setSerializeType(rpcRequest.getSerializeType());

        // 2、获取通道
        Channel channel = channelHandlerContext.channel();

        SocketAddress remoteAddress = channel.remoteAddress();
        Map<SocketAddress, RateLimiter> everyIpRateLimiter = RpcBootstrap.getInstance().getConfig().
                getEveryIpRateLimiter();

        RateLimiter rateLimiter = everyIpRateLimiter.get(remoteAddress);
        if (rateLimiter == null) {
            rateLimiter = new TokenBuketRateLimiter(100000, 100000);
            everyIpRateLimiter.put(remoteAddress, rateLimiter);
        }

        boolean allowRequest = rateLimiter.allowRequest();

        if (!allowRequest) {
            rpcResponse.setCode(ResponseCode.RATE_LIMIT.getCode());
        }
        else if (rpcRequest.getRequestType() == RequestType.HEART_BEAT.getId()) {
            rpcResponse.setCode(ResponseCode.SUCCESS_HEART_BEAT.getCode());
        } else {
            RequestPayload requestPayload = rpcRequest.getRequestPayload();
            Object res = callTargetMethod(requestPayload);
            if (log.isDebugEnabled()) {
                log.debug("请求【{}】已经在服务端完成方法调用。", rpcRequest.getRequestId());
            }

            rpcResponse.setBody(res);
            rpcResponse.setCode(ResponseCode.SUCCESS.getCode());
        }
        // 写出响应
        channel.writeAndFlush(rpcResponse);

        if (log.isDebugEnabled()) {
            log.debug("服务端完成了请求【{}】的调用，并返回了响应", rpcRequest.getRequestId());
        }

        RpcBootstrap.addOne();

    }

    private Object callTargetMethod(RequestPayload requestPayload) {
        String interfaceName = requestPayload.getInterfaceName();
        String methodName = requestPayload.getMethodName();
        Class<?>[] parametersType = requestPayload.getParametersType();
        Object[] parametersValue = requestPayload.getParametersValue();
        ServiceConfig<?> serviceConfig = RpcBootstrap.SERVICE_LIST.get(interfaceName);
        if (serviceConfig == null) {
            log.error("根据传入的接口名{}，未找到对应的实现", interfaceName);
            throw new NullPointerException();
        }
        Object refImpl = serviceConfig.getRef();

        Object returnValue;
        try {
            Class<?> aClass = refImpl.getClass();
            Method method = aClass.getMethod(methodName, parametersType);
            returnValue = method.invoke(refImpl, parametersValue);
        } catch (Exception e){
            log.error(e.getMessage(), e);
            throw new RuntimeException(e);
        }
        return returnValue;
    }
}
